package cn.ucox.web.ms.components.ftp;

import org.apache.commons.net.ftp.FTPClient;
import org.apache.commons.pool2.ObjectPool;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.PooledObjectState;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * FTP客户端连接池
 *
 * @author chenw
 * @create 2017-06-08 09:54
 * @email javacspring@gmail.com
 */
public class FTPClientPool implements ObjectPool<PooledObject<FTPClient>> {

    private static Logger logger = LoggerFactory.getLogger(FTPClient.class);

    private BlockingQueue<PooledObject<FTPClient>> mConnectionPool;
    private FtpConfig mFtpConfig;
    private FTPClientFactory mFtpClientFactory;

    public FTPClientPool(FtpConfig ftpConfig) throws Exception {
        this.mFtpConfig = ftpConfig;
        this.mFtpClientFactory = new FTPClientFactory(ftpConfig);
        this.mConnectionPool = new ArrayBlockingQueue<>(ftpConfig.getPoolSize());
        initPool();
    }

    /**
     * 初始化连接池
     *
     * @throws Exception 初始化异常
     */
    private void initPool() throws Exception {
        int count = 0;
        while (count < mFtpConfig.getPoolSize()) {
            this.addObject();
            count++;
        }
        logger.debug("Pool Size:{},Idle:{},Active:{}", mConnectionPool.size(), getNumIdle(), getNumActive());
    }

    /**
     * 从连接池中获取对象
     */
    @Override
    public PooledObject<FTPClient> borrowObject() throws Exception {
        PooledObject<FTPClient> client = mConnectionPool.take();
        if (null == client) {
            return createNewConn();
        } else {
            if (!mFtpClientFactory.validateObject(client)) {
                mFtpClientFactory.destroyObject(client);
                mConnectionPool.remove(client);
                return createNewConn();
            }
            return client;
        }
    }

    private PooledObject<FTPClient> createNewConn() throws Exception {
        if (mConnectionPool.size() < mFtpConfig.getMaxPoolSize()) {
            PooledObject<FTPClient> client = mFtpClientFactory.makeObject();
            client.use();
            client.allocate();
            mConnectionPool.offer(client);
            logger.debug("Pool Size:{},Idle:{},Active:{}", mConnectionPool.size(), getNumIdle(), getNumActive());
            return client;
        } else {
            return borrowObject();
        }
    }

    /**
     * 返还一个对象(链接)
     */
    @Override
    public void returnObject(PooledObject<FTPClient> client) throws Exception {
        if (null != client) {
            if (!mFtpClientFactory.validateObject(client)) {
                try {
                    mFtpClientFactory.destroyObject(client);
                    mConnectionPool.remove(client);
                    if (mConnectionPool.size() < mFtpConfig.getMaxPoolSize()) {
                        client = mFtpClientFactory.makeObject();
                        client.use();
                        client.allocate();
                        mConnectionPool.offer(client);
                    }
                } catch (Exception ex) {
                    logger.error("destroy ftp client error", ex);
                }
            } else {
                mFtpClientFactory.activateObject(client);
                mConnectionPool.offer(client);
            }
        }
    }

    /**
     * 移除无效的对象(FTP客户端)
     */
    @Override
    public void invalidateObject(PooledObject<FTPClient> client) throws Exception {
        mConnectionPool.remove(client);
    }

    /**
     * 增加一个新的链接，超时失效
     */
    @Override
    public void addObject() throws InterruptedException {
        mConnectionPool.offer(mFtpClientFactory.makeObject());
    }

    /**
     * 获取空闲链接数(这里暂不实现)
     */
    @Override
    public int getNumIdle() {
        return (int) mConnectionPool.stream().filter(s -> PooledObjectState.IDLE == s.getState()).count();
    }

    /**
     * 获取正在被使用的链接数
     */
    @Override
    public int getNumActive() {
        return (int) mConnectionPool.stream().filter(s -> PooledObjectState.ALLOCATED == s.getState()).count();
    }

    @Override
    public void clear() throws Exception {

    }

    /**
     * 关闭连接池
     */
    @Override
    public void close() {
        try {
            while (mConnectionPool.iterator().hasNext()) {
                mFtpClientFactory.destroyObject(mConnectionPool.take());
            }
        } catch (Exception e) {
            logger.error("close ftp client mConnectionPool failed...{}", e);
        }

    }
}
